package com.fulihui.wgj.consumer.listener;

import com.fulihui.common.constants.TopicConstants;
import com.fulihui.stater.kudu.client.KuduImpalaTemplate;
import com.fulihui.wgj.consumer.service.UserService;
import com.fulihui.wgj.facade.dto.WxUserInfoDTO;
import com.fulihui.wgj.facade.msg.UserMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author: LiuJie
 * @version: 2018/6/28 13:48
 * @description: 新用户插入
 */
@Component
@Slf4j
public class PullUserListener extends  AbstractListener<UserMsg>{
    @Autowired
    UserService userService;


    @KafkaListener(topics = TopicConstants.PULL_USER_OPENID_LIST)
    @Override
    public void listener(ConsumerRecord<String,String> record) {
        UserMsg userMsg = super.unpack(record.value(), UserMsg.class);
        this.execute(userMsg);
    }

    @Override
    public void execute(UserMsg userMsg) {
        if(null == userMsg){
            return;
        }
        // 获取用户信息
        List<WxUserInfoDTO> userInfoDTOs = userService.getUserInfo(userMsg);
        log.info("插入用户信息，{}",userInfoDTOs);
        // upsert Kudu
        userService.upsertUsers(userInfoDTOs);
    }
}
